/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package de.dnb.basics.applicationComponents;

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;

/**
 * LFU cache implementation based on http://dhruvbird.com/lfu.pdf, with some 
 * notable differences:
 * <ul>
 * <li>
 * Frequency list is stored as an array with no next/prev pointers 
 * between nodes: looping over the array should be faster and more CPU-cache 
 * friendly than
 * using an ad-hoc linked-pointers structure.
 * </li>
 * <li>
 * The max frequency is capped at the cache size to avoid creating more and
 *  more frequency list entries, and all elements residing in the max frequency 
 *  entry are re-positioned in the frequency entry linked set in order to put 
 *  most recently accessed elements ahead of less recently ones,
 * which will be collected sooner.
 * </li>
 * <li>
 * The eviction factor determines how many elements (more specifically, the 
 * percentage of) will be evicted.
 * </li>
 * </ul>
 * As a consequence, this cache runs in *amortized* O(1) time (considering the 
 * worst case of having the lowest frequency at 0 and having to evict all
 * elements).
 *
 * @author Sergio Bossa
 */
public class LFUCache<Key, Value> implements Map<Key, Value> {

	private final Map<Key, CacheNode<Key, Value>> cache;
	private final LinkedHashSet<CacheNode<Key, Value>>[] frequencyList;
	private int lowestFrequency;
	private int maxFrequency;
	//
	private final int maxCacheSize;
	private final float evictionFactor;

	public LFUCache(int maxCacheSize, float evictionFactor) {
		if (evictionFactor <= 0 || evictionFactor >= 1) {
			throw new IllegalArgumentException(
				"Eviction factor must be greater than 0 and lesser than or equal to 1");
		}
		this.cache = new HashMap<Key, CacheNode<Key, Value>>(maxCacheSize);
		this.frequencyList = new LinkedHashSet[maxCacheSize];
		this.lowestFrequency = 0;
		this.maxFrequency = maxCacheSize - 1;
		this.maxCacheSize = maxCacheSize;
		this.evictionFactor = evictionFactor;
		initFrequencyList();
	}

	@Override
	public synchronized Value put(Key k, Value v) {
		Value oldValue = null;
		CacheNode<Key, Value> currentNode = cache.get(k);
		if (currentNode == null) {
			if (cache.size() == maxCacheSize) {
				doEviction();
			}
			LinkedHashSet<CacheNode<Key, Value>> nodes = frequencyList[0];
			currentNode = new CacheNode<Key, Value>(k, v, 0);
			nodes.add(currentNode);
			cache.put(k, currentNode);
			lowestFrequency = 0;
		} else {
			oldValue = currentNode.v;
			currentNode.v = v;
		}
		return oldValue;
	}

	@Override
	public synchronized void putAll(Map<? extends Key, ? extends Value> map) {
		for (Map.Entry<? extends Key, ? extends Value> me : map.entrySet()) {
			put(me.getKey(), me.getValue());
		}
	}

	@Override
	public synchronized Value get(Object k) {
		CacheNode<Key, Value> currentNode = cache.get(k);
		if (currentNode != null) {
			int currentFrequency = currentNode.frequency;
			if (currentFrequency < maxFrequency) {
				int nextFrequency = currentFrequency + 1;
				LinkedHashSet<CacheNode<Key, Value>> currentNodes =
					frequencyList[currentFrequency];
				LinkedHashSet<CacheNode<Key, Value>> newNodes =
					frequencyList[nextFrequency];
				moveToNextFrequency(currentNode, nextFrequency,
					currentNodes, newNodes);
				cache.put((Key) k, currentNode);
				if (lowestFrequency == currentFrequency
					&& currentNodes.isEmpty()) {
					lowestFrequency = nextFrequency;
				}
			} else {
				// Hybrid with LRU: put most recently accessed ahead of others:
				LinkedHashSet<CacheNode<Key, Value>> nodes =
					frequencyList[currentFrequency];
				nodes.remove(currentNode);
				nodes.add(currentNode);
			}
			return currentNode.v;
		} else {
			return null;
		}
	}

	@Override
	public synchronized Value remove(Object k) {
		CacheNode<Key, Value> currentNode = cache.remove(k);
		if (currentNode != null) {
			LinkedHashSet<CacheNode<Key, Value>> nodes =
				frequencyList[currentNode.frequency];
			nodes.remove(currentNode);
			if (lowestFrequency == currentNode.frequency) {
				findNextLowestFrequency();
			}
			return currentNode.v;
		} else {
			return null;
		}
	}

	public int frequencyOf(Key k) {
		CacheNode<Key, Value> node = cache.get(k);
		if (node != null) {
			return node.frequency + 1;
		} else {
			return 0;
		}
	}

	@Override
	public synchronized void clear() {
		for (int i = 0; i <= maxFrequency; i++) {
			frequencyList[i].clear();
		}
		cache.clear();
		lowestFrequency = 0;
	}

	@Override
	public Set<Key> keySet() {
		return this.cache.keySet();
	}

	/**
	 * Gibt aus Implementierungsgründen null.
	 */
	@Override
	public Collection<Value> values() {
		return null; //To change body of implemented methods use File 
		//| Settings | File Templates.
	}

	/**
	 * Gibt aus Implementierungsgründen null.
	 */
	@Override
	public Set<Entry<Key, Value>> entrySet() {
		return null; //To change body of implemented methods use 
		// File | Settings | File Templates.
	}

	@Override
	public int size() {
		return cache.size();
	}

	@Override
	public boolean isEmpty() {
		return this.cache.isEmpty();
	}

	@Override
	public boolean containsKey(Object o) {
		return this.cache.containsKey(o);
	}

	/**
	 * Gibt aus Implementierungsgründen false.
	 */
	@Override
	public boolean containsValue(Object o) {
		return false; //To change body of implemented methods use File 
		// | Settings | File Templates.
	}

	private void initFrequencyList() {
		for (int i = 0; i <= maxFrequency; i++) {
			frequencyList[i] = new LinkedHashSet<CacheNode<Key, Value>>();
		}
	}

	private void doEviction() {
		int currentlyDeleted = 0;
		float target = maxCacheSize * evictionFactor;
		while (currentlyDeleted < target) {
			LinkedHashSet<CacheNode<Key, Value>> nodes =
				frequencyList[lowestFrequency];
			if (nodes.isEmpty()) {
				throw new IllegalStateException(
					"Lowest frequency constraint violated!");
			} else {
				Iterator<CacheNode<Key, Value>> it = nodes.iterator();
				while (it.hasNext() && currentlyDeleted++ < target) {
					CacheNode<Key, Value> node = it.next();
					it.remove();
					cache.remove(node.k);
				}
				if (!it.hasNext()) {
					findNextLowestFrequency();
				}
			}
		}
	}

	private void moveToNextFrequency(
		CacheNode<Key, Value> currentNode,
		int nextFrequency,
		LinkedHashSet<CacheNode<Key, Value>> currentNodes,
		LinkedHashSet<CacheNode<Key, Value>> newNodes) {
		currentNodes.remove(currentNode);
		newNodes.add(currentNode);
		currentNode.frequency = nextFrequency;
	}

	private void findNextLowestFrequency() {
		while (lowestFrequency <= maxFrequency
			&& frequencyList[lowestFrequency].isEmpty()) {
			lowestFrequency++;
		}
		if (lowestFrequency > maxFrequency) {
			lowestFrequency = 0;
		}
	}

	private static class CacheNode<Key, Value> {

		public final Key k;
		public Value v;
		public int frequency;

		public CacheNode(Key k, Value v, int frequency) {
			this.k = k;
			this.v = v;
			this.frequency = frequency;
		}

	}
}